package com.kevin.hw.zkclient;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author: kevin
 * @date: 2017/8/15
 * @description:
 */
public class MasterSelector {


    private ZkClient zkClient;

    private final static String MASTER_PATH = "/master";//需要争夺的节点名称


    private IZkDataListener dataListener;//注册节点内容变化

    private ServerInfo master;//master节点服务器

    private ServerInfo server;//其他服务器


    private boolean isRunning = false;

    private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);


    public MasterSelector(ServerInfo server,ZkClient zkClient){
        this.server = server;
        this.zkClient = zkClient;

        System.out.println("["+server+"] 去争抢master权限");
        this.dataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {

            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                chooseMaster();
            }
        };

    }


    public void start(){

        if (!isRunning){
            isRunning = true;
            zkClient.subscribeDataChanges(MASTER_PATH,dataListener); //注册节点事件
            chooseMaster();
        }

    }

    public void stop(){
        if(isRunning){
            isRunning = false;

            zkClient.subscribeDataChanges(MASTER_PATH,dataListener);
            releaseMaster();

        }

    }

    private void chooseMaster(){
        if(!isRunning){
            System.out.println("当前服务没有启动");
            return ;
        }

        try {
            zkClient.createEphemeral(MASTER_PATH,server);
            master = server;
            System.out.println(master + "已经选举为Master");
            scheduledExecutorService.schedule(() -> {
                releaseMaster();

            },2, TimeUnit.SECONDS);

        } catch (ZkNodeExistsException e) {
            ServerInfo serverInfo = zkClient.readData(MASTER_PATH,true);
            if (serverInfo == null){

                System.out.println("启动操作");
                chooseMaster();
            }else {
                master = serverInfo;
            }

        }


    }


    private void releaseMaster(){
        if (checkIsMaster()){
            zkClient.delete(MASTER_PATH);
        }


    }


    private boolean checkIsMaster(){
        ServerInfo serverInfo = zkClient.readData(MASTER_PATH);
        if (serverInfo.getMcName().equals(server.getMcName())){

            master = serverInfo;

            return true;
        }
        return false;
    }




}
